package com.by;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

public class Productor {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        //1.创建连接工厂   创建连接
        ConnectionFactory factory = new ConnectionFactory();
        factory.setVirtualHost("erp");//虚拟主机 默认是 /
        factory.setPort(5672); //连接端口号
        factory.setHost("192.168.203.130");//服务器地址 默认是localhost
        factory.setUsername("admin");
        factory.setPassword("123456");

        //2.创建连接  资源自动释放（连接、信道 关闭）
        try (Connection conn = factory.newConnection(); Channel channel = conn.createChannel()) {

            //3.启用发布确认
            channel.confirmSelect();

            //3.5.存储消息
            ConcurrentNavigableMap msgMap=new ConcurrentSkipListMap();

            //4.接收服务器确认的消息
            channel.addConfirmListener((deliveryTag, multiple) -> {
                 if(multiple){
                     //为true   批量确认   假如 2 3 5 8 当前标记为8  标记小于等于8的  都是服务器确认收到的
                     //返回所有key 小于等于deliveryTag 的集合
                     ConcurrentNavigableMap tmpMap =msgMap.headMap(deliveryTag);
                     System.out.println("tmpMap>>>>"+tmpMap.size());
                     //将消息移除
                     tmpMap.clear();
                 }else{
                     //为false  说明消息是单个确认
                     String tmp=msgMap.get(deliveryTag).toString();
                     System.out.println("消息发布成功:"+tmp);
                     //移除消息
                     msgMap.remove(deliveryTag);
                 }
            }, (deliveryTag, multiple) -> {
                System.out.println("发送失败的消息：" + msgMap.get(deliveryTag));
            });

            //4.声明队列
            String queueName = "confirm-async";
            channel.queueDeclare(queueName, false, false, false, null);


            long beign = System.currentTimeMillis();

            for (int i = 1; i <= 1000; i++) {

                //获取消息发布的序列号
                long seq=channel.getNextPublishSeqNo();
                //要发送的消息
                String msg="消息" + i;
                //将序号号消息关联想来
                msgMap.put(seq,msg);
                channel.basicPublish("", queueName, null, msg.getBytes());
            }

            System.out.println("耗费时间:" + (System.currentTimeMillis() - beign));

            System.in.read();

            System.out.println("消息发布完毕，服务器已收到");

        }


    }
}